package org.tio.cluster.kafka;

import cn.hutool.core.net.NetUtil;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.cluster.TioClusterConfig;
import org.tio.cluster.TioClusterMessageListener;
import org.tio.cluster.TioClusterTopic;
import org.tio.cluster.TioClusterVo;
import org.tio.utils.json.Json;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class KafkaTioClusterTopic implements TioClusterTopic {

    private final Logger logger = LoggerFactory.getLogger(KafkaTioClusterTopic.class);

    private KafkaProducer<String, TioClusterVo> kafkaProducer;
    private KafkaConsumer<String, TioClusterVo> kafkaConsumer;
    private final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
    private final String channel;
    private TioClusterMessageListener tioClusterMessageListener;

    public KafkaTioClusterTopic(String channel, String broker) {
        this.channel = channel;
        createProducer(broker);
        createConsumer(broker);
    }

    @Override
    public void publish(TioClusterVo tioClusterVo) {
        logger.info("publish " + Json.toJson(tioClusterVo));
        kafkaProducer.send(new ProducerRecord<>(TioClusterConfig.TIO_CLUSTER_TOPIC + channel, tioClusterVo));
    }

    @Override
    public void addMessageListener(TioClusterMessageListener tioClusterMessageListener) {
        this.tioClusterMessageListener = tioClusterMessageListener;
    }

    public void createProducer(String brokerServer) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerServer);
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.RETRIES_CONFIG, 0);
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.tio.cluster.kafka.TioClusterVoSerializer");
        kafkaProducer = new KafkaProducer(properties);
    }

    public void createConsumer(String brokerServer) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerServer);
        logger.info("local host is " + NetUtil.getLocalhostStr());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "push-connector" + NetUtil.getLocalhostStr());
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.tio.cluster.kafka.TioClusterVoDeserializer");
        kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.subscribe(Collections.singletonList(TioClusterConfig.TIO_CLUSTER_TOPIC + channel));
        executorService.execute(() -> {
            while (true) {
                ConsumerRecords<String, TioClusterVo> records = kafkaConsumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, TioClusterVo> record : records) {
                    if (tioClusterMessageListener != null) {
                        tioClusterMessageListener.onMessage(TioClusterConfig.TIO_CLUSTER_TOPIC + channel, record.value());
                    }
                }
            }
        });
    }
}
